package com.im.business.api.single;

import com.alibaba.fastjson.JSONObject;
import com.im.common.ImConstant;
import com.im.common.SessionInfo;
import com.im.common.util.SnowFlakeUtil;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import redis.clients.jedis.Jedis;

import java.util.Arrays;

/**
 * @Created leijun
 * @Date 2021/10/16 8:28 上午
 */
public class Application {

    private static SqlSessionFactory sqlSessionFactory;

    public static void main(String[] args) throws InterruptedException {
        SnowFlakeUtil snowFlakeUtil = new SnowFlakeUtil(ImConstant.DATACENTER_ID, ImConstant.MACHINE_ID);

        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaConsumer<String, String> consumer = kafkaManager.getConsumer();
        consumer.subscribe(Arrays.asList("single_message", "push_message_response"));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);
            for (ConsumerRecord record : records) {
                String topic = record.topic();
                System.out.println("【业务层】接收到kafka topic【"+topic+"】消息："+record.value());
                JSONObject message = JSONObject.parseObject(record.value().toString());
                if (topic.equals("single_message")) {
                    Long messageId = snowFlakeUtil.nextId();
                    sendMessageResponse(message, messageId);
                    if (isUserOnline(message.getString("receiverId"))) {
                        pushMessage(message, messageId);
                    } else {
                        pushOfflineMessage(message, messageId);
                    }
                } else if (topic.equals("push_message_response")) {
                    Long messageId = message.getLong("messageId");
                    updateMessageDelivered(messageId);
                }
            }
            Thread.sleep(1000);
        }
    }

    /**
     * 推送单聊的离线消息
     * @param message
     * @param messageId
     */
    private static void pushOfflineMessage(JSONObject message, Long messageId) {
        message.put("messageId", messageId);
        KafkaManager kafkaManager = KafkaManager.getInstance();
        KafkaProducer kafkaProducer = kafkaManager.getProducer();
        kafkaProducer.send(new ProducerRecord("single_offline_message", message.toJSONString()));
    }

    /**
     * 响应单聊消息
     * @param message
     * @param messageId
     */
    private static void sendMessageResponse(JSONObject message, Long messageId) {
        KafkaProducer producer = KafkaManager.getProducer();
        JSONObject messageResponse = new JSONObject();
        messageResponse.put("senderId", message.getString("senderId"));
        messageResponse.put("messageId", messageId);
        messageResponse.put("timestamp", message.getLong("timestamp"));
        ProducerRecord producerRecord = new ProducerRecord("single_message_response", messageResponse.toJSONString());
        producer.send(producerRecord);
    }

    /**
     * 推送单聊消息
     * @param message
     * @param messageId
     */
    private static void pushMessage(JSONObject message, Long messageId) {
        KafkaProducer producer = KafkaManager.getProducer();
        JSONObject messageResponse = new JSONObject();
        messageResponse.put("senderId", message.getString("senderId"));
        messageResponse.put("receiverId", message.getString("receiverId"));
        messageResponse.put("messageId", messageId);
        messageResponse.put("content", message.getString("content"));
        messageResponse.put("timestamp", message.getLong("timestamp"));
        ProducerRecord producerRecord = new ProducerRecord("push_message", messageResponse.toJSONString());
        producer.send(producerRecord);
    }

    /**
     * 更新消息状态为已送达
     * @param messageId
     */
    private static void updateMessageDelivered(Long messageId) {
        /*SqlSession sqlSession = sqlSessionFactory.openSession();
        try {
            MessageReceiveMapper messageReceiveMapper = sqlSession.getMapper(MessageReceiveMapper.class);
            MessageReceive messageReceive = new MessageReceive();
            messageReceive.setId(messageId);
            messageReceive.setIsDelivered(1);
            messageReceive.setDeliverTime(System.currentTimeMillis());
            messageReceiveMapper.updateByPrimaryKey(messageReceive);
        } finally {
            sqlSession.commit();
            sqlSession.close();
        }*/
    }

    /**
     * 用户是否在线
     * @param userId
     * @return
     */
    private static boolean isUserOnline(String userId) {
        String receiverSessionKey = "session_"+userId;
        Jedis jedis = JedisManager.getJedis();
        SessionInfo sessionInfo = JSONObject.parseObject(jedis.get(receiverSessionKey), SessionInfo.class);
        return sessionInfo == null ? false : true;
    }

}
